package com.nokia.position;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.nokia.position.PositionProto.Position;
import com.twitter.elephantbird.mapreduce.input.MultiInputFormat;

public class PositionServiceUsageDriver {

	public void printUsage() {
		System.out
				.println("Usage: hadoop jar path/to/this.jar "
						+ this.getClass()
						+ " <network type dim> <platform type dim> <protocol type dim> <operator country dim> <operator name dim> <input dir> <output dir>");

	}

	public int run(String[] args, Configuration conf) throws Exception {

		if (args.length != 7) {
			printUsage();
			return 1;
		}

		DistributedCache.addCacheFile(new Path(args[0]).toUri(), conf);
		DistributedCache.addCacheFile(new Path(args[1]).toUri(), conf);
		DistributedCache.addCacheFile(new Path(args[2]).toUri(), conf);
		DistributedCache.addCacheFile(new Path(args[3]).toUri(), conf);
		DistributedCache.addCacheFile(new Path(args[4]).toUri(), conf);

		Job job = new Job(conf);
		job.setJobName("Positioning Agg: Service Usage");

		job.setJarByClass(getClass());
		job.setMapperClass(PositionServiceUsageMapper.class);
		job.setReducerClass(PositionServiceUsageReducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(PositionServiceUsageWritable.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		// input format is same for both B64Line or block:
		MultiInputFormat.setInputFormatClass(Position.class, job);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[5]));
		FileOutputFormat.setOutputPath(job, new Path(args[6]));

		return job.waitForCompletion(true) ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		args = new GenericOptionsParser(conf, args).getRemainingArgs();
		PositionServiceUsageDriver driver = new PositionServiceUsageDriver();
		System.exit(driver.run(args, conf));
	}
}
